package com.shujia.flink.core

//导入隐式转换

import org.apache.flink.streaming.api.scala._

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
     * 创建FLink环境
     *
     */
    //获取一个环境，flink在内会自动判断是本地还是集群
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
     * 设置flink程序计算的并行度
     *
     */
    //env.setParallelism(1)


    /**
     * 设置数据从上游发送到下游的缓存的超时时间
     *
     * 默认超时时间时200毫秒
     * 当设置为0时，每一条数据发送一次---会降低flink的吞吐量
     */

    env.setBufferTimeout(200)

    /**
     * 2、读取数据
     *
     * 在linux中启动一个socket模拟流数据
     * nc -lk 8888
     *
     */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 统计单词的数量
     *
     */

    //1、将一行转换成多行
    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))

    //2、转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))

    //3、安装单词分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)

    //4、统计单词的数量，指定对哪一个列求和，指定下标，下标从0开始
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    //打印结果
    countDS.print()

    /**
     * 启动flink
     */
    env.execute()

  }

}
